package org.rcy.framework.rpc.registry.thrift;

import org.apache.thrift.TException;
import org.apache.thrift.TProcessor;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.protocol.TProtocolFactory;
import org.apache.thrift.server.TServer;
import org.apache.thrift.server.TThreadPoolServer;
import org.apache.thrift.server.TThreadPoolServer.Args;
import org.apache.thrift.server.TThreadedSelectorServer;
import org.apache.thrift.transport.TNonblockingServerSocket;
import org.apache.thrift.transport.TServerSocket;
import org.apache.thrift.transport.TServerTransport;
import org.apache.thrift.transport.TTransportFactory;
import org.apache.thrift.transport.layered.TFramedTransport;
import org.rcy.framework.rpc.config.ThriftServerConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

public class TServerThread extends Thread {
	
	private final Logger logger = LoggerFactory.getLogger(getClass());
	
	private final TServer server;
	
	public TServerThread(TProcessor processor, ThriftServerConfig serverConfig) throws Exception {
		server = threadPoolServer(processor,serverConfig);
		//server = threadedSelectorServer(processor,serverConfig);
		
		logger.info("Processor:{}",processor.getClass().getName());
		logger.info("Protocol:{}", TBinaryProtocol.class.getName());
		logger.info("Transport:{}", TFramedTransport.class.getName());
		logger.info("TServer:{}",server.getClass().getName());
	}
	
	protected TServer threadedSelectorServer(TProcessor processor, ThriftServerConfig serverConfig) throws TException {
		TNonblockingServerSocket serverTransport = new TNonblockingServerSocket(serverConfig.getPort());
		
		TProtocolFactory protocolFactory = new TBinaryProtocol.Factory(true, true);
		TTransportFactory transportFactory = new TFramedTransport.Factory();
		
		TThreadedSelectorServer.Args tArgs = new TThreadedSelectorServer.Args(serverTransport);
		tArgs.processor(processor);
		tArgs.transportFactory(transportFactory);
		tArgs.protocolFactory(protocolFactory);
		
		tArgs.selectorThreads(serverConfig.getThrift().getSelectorThreads());
		tArgs.workerThreads(serverConfig.getThrift().getWorkerThreads());
		
		return new TThreadedSelectorServer(tArgs);
	}
	protected TServer threadPoolServer(TProcessor processor, ThriftServerConfig serverConfig) throws TException{
		TServerTransport serverTransport = new TServerSocket(serverConfig.getPort());
		
		TProtocolFactory protocolFactory = new TBinaryProtocol.Factory(true, true);  
		TTransportFactory transportFactory = new TFramedTransport.Factory();
		
		TThreadPoolServer.Args tArgs = new TThreadPoolServer.Args(serverTransport);
		tArgs.processor(processor);
		tArgs.transportFactory(transportFactory);
		tArgs.protocolFactory(protocolFactory);
		
		tArgs.maxWorkerThreads(5000);
		tArgs.executorService(createThriftExecutorService(tArgs));
		
		return new TThreadPoolServer(tArgs);
	}
	
	private static ExecutorService createThriftExecutorService(Args args) {
		SynchronousQueue<Runnable> executorQueue = new SynchronousQueue<Runnable>();
		return new ThreadPoolExecutor(args.minWorkerThreads, args.maxWorkerThreads, args.stopTimeoutVal, TimeUnit.SECONDS,executorQueue, new ThriftThreadFactory());
	}

	public String getTServerInfo(){
		StringBuilder builder = new StringBuilder();
		builder.append("Protocol:"+TBinaryProtocol.class.getName());
		builder.append("Transport:"+TFramedTransport.class.getName());
		builder.append("TServer:"+server.getClass().getName());
		return builder.toString();
	}
	//启动服务
	@Override
	public void run(){
		server.serve();
	}
	
	public void shutdown(){
		server.stop();
	}
	
	
	static class ThriftThreadFactory implements ThreadFactory {
		private static final AtomicInteger poolNumber = new AtomicInteger(1);
		private final ThreadGroup group;
		private final AtomicInteger threadNumber = new AtomicInteger(1);
		private final String namePrefix;

		ThriftThreadFactory() {
			SecurityManager s = System.getSecurityManager();
			group = (s != null) ? s.getThreadGroup() : Thread.currentThread().getThreadGroup();
			namePrefix = "thrift-" + poolNumber.getAndIncrement() + "-thread-";
		}

		@Override
		public Thread newThread(Runnable r) {
			Thread t = new Thread(group, r, namePrefix + threadNumber.getAndIncrement(), 0);
			if (t.isDaemon()) {
				t.setDaemon(false);
			}
			if (t.getPriority() != Thread.NORM_PRIORITY) {
				t.setPriority(Thread.NORM_PRIORITY);
			}
			return t;
		}
	}
}